﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Caching;
using System.Runtime.CompilerServices;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Fasterflect;
using Kugar.Core.BaseStruct;
using Kugar.Core.ExtMethod;
using Kugar.Tool.RemoteEventBus.DTOPackages;
using Kugar.Tool.RemoteEventBus.Helpers;
using Kugar.Tool.RemoteEventBus.Subscibers;
using Kugar.Tool.RemoteEventBus.Triggers;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Framing;

namespace Kugar.Tool.RemoteEventBus
{
    public class RemoteEventBus : IRemoteEventBus //:IRemoteEventBus
    {
        private const string _brocastExchange = "RemoteEventBus_Brocast";
        private const string _asyncExchange = "RemoteEventBus_Async";
        private const string _syncExchange = "RemoteEventBus_Sync";

        private const string _nodeOnlieRouteKey = "Node_Online";
        private const string _nodeOfflineRouteKey = "Node_Offline";
        private const string _nodeSubscibeRouteKey = "Node_EventSubscibe";
        private const string _nodeUnSubscibeRouteKey = "Node_EventUnSubscibe";
        private const string _nodeNewEventOnineRouteKey = "Node_NewEvent_Online";
        private const string _nodeKeepAliveRouteKey = "Node_KeepAlive";

        private static RemoteEventBus _defaultValue = new RemoteEventBus();


        //该路由指向当前节点,用于当本节点上线时,其他节点向本节点发送补充的Welcome订阅消息
        //private string _nodePrivateNodeOnlineRouteKey = "Node_Online_Welcome_{0}";

        private string _nodePrivateSubscibeRouteKey = "Node_Private_EventSubscibe_{0}";


        private IConnection _conn = null;
        private string _serviceID = "";
        private Dictionary<string, object> _cacheEventTrigger = new Dictionary<string, object>(StringComparer.CurrentCultureIgnoreCase);
        private Dictionary<string, IEventSubsciber> _cacheEventSubscriber = new Dictionary<string, IEventSubsciber>(StringComparer.CurrentCultureIgnoreCase);
        private bool _isInit = false;
        private IModel _channel = null;
        private string _queueName = ""; //节点队列名称,用于接收节点上下线,消息订阅退订等消息
        private EventingBasicConsumer _consumer = null;
        private MemoryCache _cacheServiceAlive = new MemoryCache("RemoteEventBus_Service");
        private ReaderWriterLockSlim _triggerLocker = new ReaderWriterLockSlim();
        private ReaderWriterLockSlim _subscriberLocker = new ReaderWriterLockSlim();


        /// <summary>
        /// 初始化当前实例,传入RabbitMQ的链接,当前服务的名称ID
        /// </summary>
        /// <param name="conn"></param>
        /// <param name="serviceID"></param>
        public void Initialize(IConnection conn, string serviceID)
        {
            if (conn == null || !conn.IsOpen)
            {
                throw new ArgumentException("conn必须为以打开的连接");
            }

            if (_isInit)
            {
                return;
            }

            _serviceID = serviceID;

            _nodePrivateSubscibeRouteKey = string.Format(_nodePrivateSubscibeRouteKey, _serviceID);
            //_nodePrivateNodeOnlineRouteKey = string.Format(_nodePrivateNodeOnlineRouteKey, _serviceID);

            _conn = conn;

            _channel = _conn.CreateModel();

            _queueName = $"Node_Queue_{serviceID}";

            _channel.ExchangeDeclare(_asyncExchange, "fanout", false, false, null);
            _channel.ExchangeDeclare(_syncExchange, "fanout", false, false, null);
            _channel.ExchangeDeclare(_brocastExchange, "fanout", false, false, null);

            var brocastQueue = _channel.QueueDeclare(_queueName, false, false, true, null);

            _channel.QueueBind(brocastQueue.QueueName, _brocastExchange, _nodeOnlieRouteKey);
            _channel.QueueBind(brocastQueue.QueueName, _brocastExchange, _nodeOfflineRouteKey);
            _channel.QueueBind(brocastQueue.QueueName, _brocastExchange, _nodeSubscibeRouteKey);
            _channel.QueueBind(brocastQueue.QueueName, _brocastExchange, _nodeUnSubscibeRouteKey);
            //_channel.QueueBind(brocastQueue.QueueName, _brocastExchange, _nodePrivateNodeOnlineRouteKey);
            _channel.QueueBind(brocastQueue.QueueName, _brocastExchange, _nodePrivateSubscibeRouteKey);
            _channel.QueueBind(brocastQueue.QueueName, _brocastExchange, _nodeNewEventOnineRouteKey);

            _consumer = new EventingBasicConsumer(_channel);
            _consumer.Received += Consumer_Received;

            _channel.BasicConsume(_queueName, false, _consumer);

            _isInit = true;


            //初始化结束后,发送节点上线消息
            Push_currentNodeOnline();
            
            TimerPool.Default.Add(60000, keepNodeAlive);
            
        }

        private void Subsciver_AsyncEventRaised(object sender, AsyncEventRaisedEventArgs e)
        {
            try
            {
                this.AsyncEventRaised?.Invoke(sender, e);
            }
            catch (Exception exception)
            {

            }
        }

        private void Subsciver_SyncEventRaised(object sender, SyncEventRaisedEventArgs e)
        {
            try
            {
                this.SyncEventRaised?.Invoke(sender, e);
            }
            catch (Exception exception)
            {

            }
        }

        ///// <summary>
        ///// 在程序初始化完成后,调用该函数,调用后,将会完成整个类的初始化过程,并将向外发送节点上线通知,建议在调用该函数前,将所需的事件订阅器都创建完,避免每次创建时都去通知其他节点
        ///// </summary>
        //public void InitCompleted()
        //{
        //    _isInit = true;

        //    Push_currentNodeOnline();
        //}

        public AsyncEventSubsciber<T> GetAsyncEventSubsciber<T>(string eventName) //where T : EventArgs
        {
            _subscriberLocker.EnterUpgradeableReadLock();

            try
            {
                IEventSubsciber value;

                if (_cacheEventSubscriber.TryGetValue(eventName, out value))
                {
                    return (AsyncEventSubsciber<T>)value;
                }
                else
                {
                    _subscriberLocker.EnterWriteLock();

                    value = createAsyncSubsciber<T>(eventName);

                    _cacheEventSubscriber.Add(eventName, value);

                    var subsciver= (AsyncEventSubsciber<T>)value;

                    subsciver.EventRaised += Subsciver_AsyncEventRaised; ;

                    return subsciver;
                }
            }
            catch (Exception)
            {
                throw;
            }
            finally
            {
                if (_subscriberLocker.IsWriteLockHeld)
                {
                    _subscriberLocker.ExitWriteLock();
                }

                _subscriberLocker.ExitUpgradeableReadLock();
            }


            //var item = (AsyncEventSubsciber<T>)_cacheEventSubscriber.GetOrAdd(eventName, createAsyncSubsciber<T>);

            //return item;
        }
        
        public AsyncEventTrigger<T> GetAsyncEventTrigger<T>(string eventName) //where T : EventArgs
        {
            _triggerLocker.EnterUpgradeableReadLock();

            try
            {
                object value;
                if (_cacheEventTrigger.TryGetValue(eventName, out value))
                {
                    return (AsyncEventTrigger<T>)value;
                }
                else
                {
                    _triggerLocker.EnterWriteLock();

                    value = createAsyncTrigger<T>(eventName);

                    _cacheEventTrigger.Add(eventName, value);

                    if (_isInit)
                    {
                        Push_currentNewEventOnline(eventName);
                    }



                    var trigger= (AsyncEventTrigger<T>)value;

                    trigger.TriggerEvent += Trigger_AsyncTriggerEvent; ;
                    trigger.ClientHandledEvent += Trigger_AsyncClientHandledEvent; ;

                    return trigger;
                }
            }
            catch (Exception)
            {
                throw;
            }
            finally
            {
                if (_triggerLocker.IsWriteLockHeld)
                {
                    _triggerLocker.ExitWriteLock();
                }

                _triggerLocker.ExitUpgradeableReadLock();
            }

            //return (AsyncEventTrigger<T>)_cacheEventTrigger.GetOrAdd(eventName, createAsyncTrigger<T>);
        }



        public SyncEventSubsecriber<T> GetSyncEventSubsecriber<T>(string eventName) where T : SyncRaiseEventArgs
        {
            _subscriberLocker.EnterUpgradeableReadLock();

            try
            {
                IEventSubsciber value;

                if (_cacheEventSubscriber.TryGetValue(eventName, out value))
                {
                    return (SyncEventSubsecriber<T>)value;
                }
                else
                {
                    _subscriberLocker.EnterWriteLock();

                    value = createSyncSubsciber<T>(eventName);

                    _cacheEventSubscriber.Add(eventName, value);

                    //return (SyncEventSubsecriber<T>)value;

                    var subsciver = (SyncEventSubsecriber<T>)value;

                    subsciver.EventRaised += Subsciver_SyncEventRaised; ;

                    return subsciver;
                }
            }
            catch (Exception)
            {
                throw;
            }
            finally
            {
                if (_subscriberLocker.IsWriteLockHeld)
                {
                    _subscriberLocker.ExitWriteLock();
                }

                _subscriberLocker.ExitUpgradeableReadLock();
            }

            //return (SyncEventSubsecriber<T>)_cacheEventSubscriber.GetOrAdd(eventName, createSyncSubsciber<T>);
        }
        
        public SyncEventTrigger<T> GetSyncEventTrigger<T>(string eventName) where T : EventArgs
        {
            _triggerLocker.EnterUpgradeableReadLock();

            try
            {
                object value;
                if (_cacheEventTrigger.TryGetValue(eventName, out value))
                {
                    return (SyncEventTrigger<T>)value;
                }
                else
                {
                    _triggerLocker.EnterWriteLock();

                    value = createSyncTrigger<T>(eventName);

                    _cacheEventTrigger.Add(eventName, value);

                    if (_isInit)
                    {
                        Push_currentNewEventOnline(eventName);
                    }

                    var trigger= (SyncEventTrigger<T>)value;

                    trigger.TriggerEvent += Trigger_SyncTriggerEvent; ;
                    trigger.ClientHandledEvent += Trigger_SyncClientHandledEvent; ;

                    return trigger;
                }
            }
            catch (Exception)
            {
                throw;
            }
            finally
            {
                if (_triggerLocker.IsWriteLockHeld)
                {
                    _triggerLocker.ExitWriteLock();
                }

                _triggerLocker.ExitUpgradeableReadLock();
            }


            //return (SyncEventTrigger<T>)_cacheEventTrigger.GetOrAdd(eventName, createSyncTrigger<T>);
        }



        public string ServiceID => _serviceID;

        public IConnection Connection => _conn;

        public IEnumerable<IEventTrigger> Triggers
        {
            get
            {
                _triggerLocker.EnterReadLock();

                foreach (var trigger in _cacheEventTrigger)
                {
                    yield return (IEventTrigger)trigger.Value;
                }

                _triggerLocker.ExitReadLock();
            }
        }

        public IEnumerable<IEventSubsciber> Subscribers
        {
            get
            {
                _subscriberLocker.EnterReadLock();

                foreach (var subsecriber in _cacheEventSubscriber)
                {
                    yield return (IEventSubsciber)subsecriber.Value;
                }

                _subscriberLocker.ExitReadLock();
            }
        }



        public void Close()
        {
            Push_currentNodeOffline();
            
            //清理触发器
            _triggerLocker.EnterWriteLock();

            foreach (var trigger in _cacheEventTrigger)
            {
                try
                {
                    ((IEventTriggerSubscriberHandler)trigger.Value).ClearSubscriber();
                    ((IEventTriggerSubscriberHandler)trigger.Value).Close();
                }
                catch (Exception)
                {
                }
            }

            _cacheEventTrigger.Clear();

            _triggerLocker.ExitWriteLock();


            //清理订阅器
            _subscriberLocker.EnterWriteLock();

            foreach (var sub in _cacheEventSubscriber)
            {
                try
                {
                    ((IEventSubsciber)sub.Value).Close();
                }
                catch (Exception)
                {
                }
                
            }

            _cacheEventSubscriber.Clear();

            _subscriberLocker.ExitWriteLock();

            
            

            var serviceIDs = _cacheServiceAlive.Select(x => x.Key);

            foreach (var serviceID in serviceIDs)
            {
                _cacheServiceAlive.Remove(serviceID);
            }

            _channel.Close();
        }

        public event EventHandler<NodeOfflineEventArgs> NodeOffline;
        public event EventHandler<NodeOnlineEventArgs> NodeOnline;
        public event EventHandler<NodeEventSubscibeEventArgs> NodeEventSubscibe;

        public event EventHandler<SyncEventRaisedEventArgs> SyncEventRaised;
        public event EventHandler<AsyncEventRaisedEventArgs> AsyncEventRaised;

        public event EventHandler<ClientEventHandeEventArgs> SyncEventClientHandled;
        public event EventHandler<ClientEventHandeEventArgs> AsyncEventClientHandled;

        public event EventHandler<EventTriggerEventArgs>  SyncEventTrigger;
        public event EventHandler<EventTriggerEventArgs> AsyncEventTrigger;
        
        public static RemoteEventBus Default => _defaultValue;

        private void Consumer_Received(object sender, BasicDeliverEventArgs e)
        {
            var routeKey = e.RoutingKey;

            if (e.BasicProperties.CorrelationId.CompareTo(_serviceID, true))
            {
                _channel.BasicAck(e.DeliveryTag, false);
                return;
            }

            try
            {
                if (routeKey == _nodeOnlieRouteKey) //当接收到节点上线时,触发该函数
                {
                    Receive_remoteNodeOnline(e.BasicProperties.CorrelationId, e.BasicProperties.ReplyTo, e.Body);
                }
                else if (routeKey == _nodeOfflineRouteKey)
                {
                    Receive_remoteNodeOffline(e.BasicProperties.CorrelationId);
                }
                else if (routeKey == _nodeSubscibeRouteKey ||
                         routeKey == _nodePrivateSubscibeRouteKey ||
                         routeKey==_nodeKeepAliveRouteKey
                         )  //当订阅或者补充订阅消息时,触发该事件
                {

                    var args = JsonConvert.DeserializeObject<NodeSubscibeEventPackage>(Encoding.UTF8.GetString(e.Body));

                    //Console.WriteLine($"订阅事件={JsonConvert.SerializeObject(args)}");

                    Receive_remoteEventSubscibe(args.SubsciberServiceID, args.EventNames);
                }
                else if (routeKey == _nodeNewEventOnineRouteKey) //当其他节点有新事件上线时,触发该函数
                {
                    var args = JsonConvert.DeserializeObject<NodeNewEventOnlinePackage>(Encoding.UTF8.GetString(e.Body));

                    Receive_remoteNodeNewEventOnline(e.BasicProperties.CorrelationId, e.BasicProperties.ReplyTo, args.EventName);
                }
                else if (routeKey == _nodeUnSubscibeRouteKey)
                {
                    var args = JsonConvert.DeserializeObject<NodeUnsubscibeEventPackage>(Encoding.UTF8.GetString(e.Body));
                    Receive_remoteEventUnSubscibe(args.SubsciberServiceID, args.EventNames);
                }
            }
            catch (Exception)
            {
                //TODO:触发日志记录
            }
            finally
            {
                _channel.BasicAck(e.DeliveryTag, false);
                
                if (_cacheServiceAlive.Contains(e.BasicProperties.CorrelationId))
                {
                    _cacheServiceAlive[e.BasicProperties.CorrelationId] = DateTime.Now;
                }
            }

        }

        /// <summary>
        /// 推送一条消息,表明当前节点的活动状态
        /// </summary>
        /// <param name="state"></param>
        private void keepNodeAlive(object state)
        {
            Push_currentEventSubscibe(_nodeKeepAliveRouteKey,_cacheEventSubscriber.Keys.ToArray());
        }

        /// <summary>
        /// 当前节点上线消息,发送上线消息到其他接收器中
        /// </summary>
        private void Push_currentNodeOnline()
        {
            _channel.BasicPublish(
                _brocastExchange,
                _nodeOnlieRouteKey,
                _serviceID,
                _nodePrivateSubscibeRouteKey,
                new NodeOnlinePackage()
                {
                    ServiceID = _serviceID,
                    SubscibeEventNames = _cacheEventSubscriber.Keys.ToArrayEx(),
                    SupportEventNames = _cacheEventTrigger.Keys.ToArrayEx()
                }
                );

            //_channel.BasicPublish(_brocastExchange,
            //    _nodeOnlieRouteKey,
            //    true,
            //    new BasicProperties()
            //    {
            //        CorrelationId = _serviceID,
            //        AppId = "RemoteEventBus",
            //        Expiration = "15000",
            //        MessageId = Guid.NewGuid().ToString(),
            //        ReplyTo = _nodePrivateSubscibeRouteKey,
            //    },
            //    Encoding.UTF8.GetBytes(
            //        JsonConvert.SerializeObject(new NodeOnlinePackage()
            //        {
            //            ServiceID = _serviceID,
            //            SubscibeEventNames = _cacheEventSubscriber.Keys.ToArrayEx(),
            //            SupportEventNames = _cacheEventTrigger.Keys.ToArrayEx()
            //        })
            //        )
            //    );
        }

        /// <summary>
        /// 远程其他节点上线消息,接收到该消息,则表示有一个节点已上线
        /// </summary>
        private void Receive_remoteNodeOnline(string serviceID, string relyRouteKey, byte[] data)
        {
            var package = JsonConvert.DeserializeObject<NodeOnlinePackage>(Encoding.UTF8.GetString(data));

            Receive_remoteEventSubscibe(serviceID, package.SubscibeEventNames);

            _cacheServiceAlive.Set(serviceID, DateTime.Now, new CacheItemPolicy()
            {
                SlidingExpiration = new TimeSpan(0, 3, 0),
                RemovedCallback = serviceTimeoutOffline
            });



            if (_cacheEventSubscriber.Count > 0 && package.SupportEventNames.HasData())
            {
                bool isNeedSubscribe = false;

                _subscriberLocker.EnterReadLock();

                try
                {
                    //var keys = _cacheEventSubscriber.Keys.ToArrayEx();

                    foreach (var pair in _cacheEventSubscriber)
                    {
                        isNeedSubscribe = package.SupportEventNames.Contains(pair.Key);

                        if (isNeedSubscribe)
                        {
                            break;
                        }
                    }
                }
                catch (Exception)
                {

                    throw;
                }
                finally
                {
                    _subscriberLocker.ExitReadLock();
                }

                if (isNeedSubscribe)
                {
                    Push_currentEventSubscibe(relyRouteKey, package.SupportEventNames);
                }
            }

            if (NodeOnline != null)
            {
                NodeOnline(this, new NodeOnlineEventArgs(serviceID));
            }

        }

        /// <summary>
        /// 向外发送一条事件订阅的消息,通知其他节点,本节点要订阅某些事件
        /// </summary>
        private void Push_currentEventSubscibe(string[] eventNames)
        {


            Push_currentEventSubscibe(_nodeSubscibeRouteKey, eventNames);

            //_channel.BasicPublish(_brocastExchange,
            //    _nodeSubscibeRouteKey,
            //    true,
            //    new BasicProperties()
            //    {
            //        CorrelationId = _serviceID,
            //        AppId = "RemoteEventBus",
            //        Expiration = "15000",
            //        MessageId = Guid.NewGuid().ToString(),
            //        ReplyTo = _queueName,
            //    },
            //    Encoding.UTF8.GetBytes(
            //        JsonConvert.SerializeObject(new NodeSubscibeEventPackage()
            //        {
            //            EventNames = eventNames,
            //            SubsciberServiceID = this._serviceID
            //        })
            //        )
            //    );
        }

        private void Push_currentEventSubscibe(string routeKey, string[] eventNames)
        {


            _channel.BasicPublish(_brocastExchange,
                routeKey,
                _serviceID,
                _queueName,
                new NodeSubscibeEventPackage()
                {
                    EventNames = eventNames,
                    SubsciberServiceID = this._serviceID
                },
                messageID: Guid.NewGuid().ToString()
                );

            //_channel.BasicPublish(_brocastExchange,
            //    routeKey,//_nodeSubscibeRouteKey,
            //    true,
            //    new BasicProperties()
            //    {
            //        CorrelationId = _serviceID,
            //        AppId = "RemoteEventBus",
            //        Expiration = "15000",
            //        MessageId = Guid.NewGuid().ToString(),
            //        ReplyTo = _queueName,
            //    },
            //    Encoding.UTF8.GetBytes(
            //        JsonConvert.SerializeObject(new NodeSubscibeEventPackage()
            //        {
            //            EventNames = eventNames,
            //            SubsciberServiceID = this._serviceID
            //        })
            //        )
            //    );
        }

        /// <summary>
        /// 接收到其他节点要订阅某些事件的消息,
        /// </summary>
        private void Receive_remoteEventSubscibe(string serviceID, string[] eventNames)
        {
            //Console.WriteLine($"远程订阅事件={eventNames.JoinToString()}");

            foreach (var eventName in eventNames)
            {
                if (!_cacheEventTrigger.ContainsKey(eventName))
                {
                    continue;
                }

                var trigger = (IEventTriggerSubscriberHandler)_cacheEventTrigger.TryGetValue(eventName);

                trigger.AddSubscriber(serviceID);
            }

            if (NodeEventSubscibe!=null)
            {
                NodeEventSubscibe(this,new NodeEventSubscibeEventArgs(serviceID,eventNames));
            }
        }

        /// <summary>
        /// 向外发送一条事件退订的消息,通知其他节点,本节点要退订某些事件
        /// </summary>
        private void Push_currentEventUnSubscibe(string[] eventNames)
        {
            _channel.BasicPublish(
                _brocastExchange,
                _nodeUnSubscibeRouteKey,
                _serviceID,
                _queueName,
                new NodeSubscibeEventPackage()
                {
                    EventNames = eventNames,
                    SubsciberServiceID = this._serviceID
                }
                );

            //_channel.BasicPublish(_brocastExchange,
            //    _nodeUnSubscibeRouteKey,
            //    true,
            //    new BasicProperties()
            //    {
            //        CorrelationId = _serviceID,
            //        AppId = "RemoteEventBus",
            //        Expiration = "15000",
            //        MessageId = Guid.NewGuid().ToString(),
            //        ReplyTo = _queueName,
            //    },
            //    Encoding.UTF8.GetBytes(
            //        JsonConvert.SerializeObject(new NodeSubscibeEventPackage()
            //        {
            //            EventNames = eventNames,
            //            SubsciberServiceID = this._serviceID
            //        })
            //        )
            //    );
        }

        /// <summary>
        /// 接收到其他节点要退订某些事件的消息,
        /// </summary>
        private void Receive_remoteEventUnSubscibe(string serviceID, string[] eventNames)
        {
            foreach (var eventName in eventNames)
            {
                if (!_cacheEventTrigger.ContainsKey(eventName))
                {
                    continue;
                }

                var trigger = (IEventTriggerSubscriberHandler)_cacheEventTrigger.TryGetValue(eventName);

                trigger.RemoveSubscriber(serviceID);
            }
        }

        /// <summary>
        /// 向外发送一条消息,表明当前节点即将离线
        /// </summary>
        private void Push_currentNodeOffline()
        {
            _channel.BasicPublish(
                _brocastExchange,
                _nodeOfflineRouteKey,
                _serviceID,
                _queueName,
                null
                );

            //_channel.BasicPublish(_brocastExchange,
            //    _nodeOfflineRouteKey,
            //    true,
            //    new BasicProperties()
            //    {
            //        CorrelationId = _serviceID,
            //        AppId = "RemoteEventBus",
            //        Expiration = "15000",
            //        MessageId = Guid.NewGuid().ToString(),
            //        ReplyTo = _queueName,
            //    },
            //    null
            //    );
        }

        /// <summary>
        /// 接收到其他节点离线的消息
        /// </summary>
        /// <param name="serviceID"></param>
        private void Receive_remoteNodeOffline(string serviceID)
        {
            foreach (var trigger in _cacheEventTrigger)
            {
                ((IEventTriggerSubscriberHandler)trigger.Value).RemoveSubscriber(serviceID);
            }

            _cacheServiceAlive.Remove(serviceID);

            if (NodeOffline != null)
            {
                NodeOffline(this, new NodeOfflineEventArgs(serviceID));
            }
        }

        private void Push_currentNewEventOnline(string eventName)
        {
            _channel.BasicPublish(
                _brocastExchange,
                _nodeNewEventOnineRouteKey,
                _serviceID,
                _nodePrivateSubscibeRouteKey,
                new NodeNewEventOnlinePackage()
                {
                    ServiceID = _serviceID,
                    EventName = eventName
                }

                );

            //_channel.BasicPublish(
            //        _brocastExchange,
            //        _nodeNewEventOnineRouteKey,
            //        true,
            //        new BasicProperties()
            //        {
            //            Expiration = "15000",
            //            CorrelationId = _serviceID,
            //            ReplyTo = _nodePrivateSubscibeRouteKey,
            //            MessageId = Guid.NewGuid().ToString()
            //        },
            //        Encoding.UTF8.GetBytes(
            //            JsonConvert.SerializeObject(new NodeNewEventOnlinePackage()
            //            {
            //                ServiceID = _serviceID,
            //                EventName = eventName
            //            })
            //            )
            //        );
        }

        private void Receive_remoteNodeNewEventOnline(string serviceID, string relyRouteKey, string eventName)
        {
            if (!_cacheEventSubscriber.ContainsKey(eventName))
            {
                return;
            }

            //Console.WriteLine($"远程新增事件={eventName}");

            Push_currentEventSubscibe(relyRouteKey, new[] { eventName });

            //_channel.BasicPublish(
            //    _brocastExchange,
            //    relyRouteKey,
            //    _serviceID,
            //    _nodePrivateSubscibeRouteKey,
            //    new NodeSubscibeEventPackage()
            //        {
            //            EventNames = new string[] { eventName },
            //            SubsciberServiceID = _serviceID
            //        }
            //    );

            //_channel.BasicPublish(_brocastExchange,
            //    relyRouteKey,
            //    true,
            //    new BasicProperties()
            //    {
            //        Expiration = "15000",
            //        CorrelationId = _serviceID,
            //        AppId = "RemoteEventBus",
            //    },
            //    Encoding.UTF8.GetBytes(
            //        JsonConvert.SerializeObject(new NodeSubscibeEventPackage()
            //        {
            //            EventNames = new string[] { eventName },
            //            SubsciberServiceID = _serviceID
            //        })
            //        )
            //    );
        }


        private IEventSubsciber createAsyncSubsciber<T>(string eventName) //where T : EventArgs
        {
            if (_isInit)
            {
                Push_currentEventSubscibe(new[] { eventName });
            }

            return new AsyncEventSubsciber<T>(this, eventName);
        }

        private object createAsyncTrigger<T>(string eventName) //where T : EventArgs
        {
            return new AsyncEventTrigger<T>(this, eventName);
        }

        private IEventSubsciber createSyncSubsciber<T>(string eventName) where T : SyncRaiseEventArgs
        {
            if (_isInit)
            {
                Push_currentEventSubscibe(new[] { eventName });
            }

            return new SyncEventSubsecriber<T>(this, eventName);
        }

        private object createSyncTrigger<T>(string eventName) where T : EventArgs
        {
            return new SyncEventTrigger<T>(this, eventName);
        }

        private void serviceTimeoutOffline(CacheEntryRemovedArguments arguments)
        {
            var serviceID = arguments.CacheItem.Key;
            
            Receive_remoteNodeOffline(serviceID);
        }

        private void Trigger_SyncClientHandledEvent(object sender, ClientEventHandeEventArgs e)
        {
            try
            {
                this.SyncEventClientHandled?.Invoke(sender, e);
            }
            catch (Exception exception)
            {

            }
        }

        private void Trigger_SyncTriggerEvent(object sender, EventTriggerEventArgs e)
        {
            try
            {
                this.SyncEventTrigger?.Invoke(sender, e);
            }
            catch (Exception exception)
            {

            }
        }

        private void Trigger_AsyncClientHandledEvent(object sender, ClientEventHandeEventArgs e)
        {
            try
            {
                this.AsyncEventClientHandled?.Invoke(sender, e);
            }
            catch (Exception exception)
            {

            }

        }

        private void Trigger_AsyncTriggerEvent(object sender, EventTriggerEventArgs e)
        {
            try
            {
                this.AsyncEventTrigger?.Invoke(sender, e);
            }
            catch (Exception exception)
            {
            }
        }
    }
    

    //public class RemoteEventBus : IRemoteEventBus
    //{
    //    private static Lazy<RemoteEventBus> _defaultValue = new Lazy<RemoteEventBus>(()=>new RemoteEventBus());

    //    private IConnection _conn = null;
    //    private IModel _channel = null;
    //    private string _serviceID = "";
    //    private bool _isInit = false;
    //    private string _eventQueue = "";
    //    private MemoryCache _callback = new MemoryCache("SyncCallbackCache");
    //    private Dictionary<string, MemoryCache> _eventBus = new Dictionary<string, MemoryCache>();
    //    private HashSet<string> _currentSubscibeEvents = new HashSet<string>();
    //    private Dictionary<string, object> _cacheEventSubsciber = new Dictionary<string, object>();
    //    private EventingBasicConsumer _consumer = null;
    //    private ReaderWriterLockSlim _subscriberLocker=new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);

    //    public static RemoteEventBus Default => _defaultValue.Value;

    //    /*
    //     * 1.ServiceID  节点ID,用于标识当前节点,最好根据服务器不同,修改该节点ID,也可以使用服务器名,尽量让该ID唯一
    //     * 
    //     * 2.Exchange:分为两个,,RemoteEvent_Brocast  用于发布广播消息,比如节点上线的消息,节点订阅/退订事件的消息,节点离线消息(fanout 模式)
    //     *                      RemoteEvent_EventBus 用于实际事件触发时的消息发送和接收处理
    //     * 
    //     * 3.QueueName 队列名的规则是:{ServiceID}_Brocast  该队列用户接收RemoteEvent_Brocast发送的广播消息
    //     *                            {ServiceID}_EventChannel  该队列用于接收实际事件触发消息
    //     *                            
    //     * 4.RouteKey  路由名称 RemoteEvent_EventBus 的路由规则是 {EventName}_Sync_Trigger  表示一个同步事件的触发消息
    //     *                                                       {EventName}_Sync_Rely  表示一个同步事件的回复消息
    //     *                                                       {EventName}_Async_Trigger 表示一个异步事件的触发消息(无需回复)
    //     *                                                       Queue_{ServiceID} 节点特定的队列
    //     *                                                       
    //     *                      RemoteEvent_Brocast 的路由规则是 Node_Online 表示一个新的节点上线
    //     *                                                       EventSubscibe 表示节点订阅一个指定名称的事件(各个节点根据自身提供的事件绑定队列的路由)
    //     *                                                       EventUnSubscibe 表示节点退订一个指定名称的事件(各个节点根据自身提供的事件绑定队列的路由)
    //     *                                                       EventTriggerOnline 表示节点增加一个触发的事件
    //     *                                                       Queue_{ServiceID} 节点特定的队列
    //     *                                                       
    //     *  A节点上线后,广播一条 Node_Online的消息,接收到该消息的其他节点,向A节点对应的队列推送所有已订阅的事件,
    //     *  A节点上线接收到该消息之后,根据自身的的事件列表,处理订阅列表
    //     */

    //    /// <summary>
    //    /// 初始化远程事件总线,调用该函数之后,其他服务器节点会触发NodeOnlie事件
    //    /// </summary>
    //    /// <param name="conn">RabbitMQ连接对象</param>
    //    /// <param name="serviceID">当前节点ID,尽量保证该值唯一并且当前节点的值不变,方便查看节点的消息队列</param>
    //    public void Initialize(IConnection conn, string serviceID)
    //    {
    //        if (conn==null || !conn.IsOpen)
    //        {
    //            throw new ArgumentException("conn必须为以打开的连接");
    //        }

    //        if (_isInit)
    //        {
    //            return;
    //        }

    //        _serviceID = serviceID;

    //        _isInit = true;

    //        _conn = conn;

    //        _channel = _conn.CreateModel();

    //        _channel.BasicReturn += channel_BasicReturn;

    //        _channel.ExchangeDeclare("RemoteEvent_Brocast", "fanout", false, false, null);
    //        _channel.ExchangeDeclare("RemoteEvent_EventBus", "direct", false, false, null);
    //        _eventQueue = $"Queue_{serviceID}";

    //        var brocastQueue = _channel.QueueDeclare(_eventQueue, true, false, true, null);

    //        _channel.QueueBind(brocastQueue.QueueName, "RemoteEvent_Brocast", "NodeOnline");
    //        _channel.QueueBind(brocastQueue.QueueName, "RemoteEvent_Brocast", "NodeSubscibeEvent");
    //        _channel.QueueBind(brocastQueue.QueueName, "RemoteEvent_EventBus", $"Queue_{serviceID}");

    //        _consumer = new EventingBasicConsumer(_channel);

    //        _consumer.Received += Consumer_Received;

    //        _channel.BasicConsume(brocastQueue.QueueName, false, _consumer);

    //        _channel.BasicConsume(_eventQueue, false, _consumer);

    //        //初始化结束后,发送节点上线消息
    //        currentNodeOnline();

    //        TimerPool.Default.Add(30000, keepNodeAlive, null);
    //    }

    //    public string ServiceID => _serviceID;

    //    public bool IsDevelopNode { get; } = false;

    //    /// <summary>
    //    /// 触发一个同步事件
    //    /// </summary>
    //    /// <param name="eventName"></param>
    //    /// <param name="eventData"></param>
    //    public async Task<JObject[]> TriggerSyncEvent<T>(string eventName, T eventData) where T : SyncEventArgs
    //    {
    //        return await syncEventTrigger(eventName, eventData);
    //    }

    //    /// <summary>
    //    /// 触发一个异步事件
    //    /// </summary>
    //    /// <typeparam name="T"></typeparam>
    //    /// <param name="eventName"></param>
    //    /// <param name="eventData"></param>
    //    public void TriggerAsyncEvent<T>(string eventName, T eventData) //where T : EventArgs
    //    {
    //        asyncEventTrigger(eventName, eventData);
    //    }



    //    /// <summary>
    //    /// 创建一个同步事件
    //    /// </summary>
    //    /// <param name="eventName"></param>
    //    public void CreateSyncEvent(string eventName)
    //    {
    //        createEventTrigger(new[] { eventName }, false);
    //    }

    //    /// <summary>
    //    /// 创建一个同步事件,并返回一个触发器
    //    /// </summary>
    //    /// <typeparam name="TEventArgs"></typeparam>
    //    /// <param name="eventName"></param>
    //    /// <returns></returns>
    //    public ISyncTrigger<TEventArgs> CreateSyncEvent<TEventArgs>(string eventName) where TEventArgs : SyncEventArgs
    //    {
    //        CreateSyncEvent(eventName);

    //        return new SyncTrigger<TEventArgs>(this,eventName);
    //    }

    //    /// <summary>
    //    /// 创建一个异步事件
    //    /// </summary>
    //    /// <param name="eventName"></param>
    //    public void CreateAsyncEvent(string eventName)
    //    {
    //        createEventTrigger(new[] { eventName }, true);
    //    }

    //    /// <summary>
    //    /// 创建一个异步事件,并返回触发器
    //    /// </summary>
    //    /// <typeparam name="TEventArgs">事件发的参数</typeparam>
    //    /// <param name="eventName">事件名</param>
    //    /// <returns></returns>
    //    public IAsyncTrigger<TEventArgs> CreateAsyncEvent<TEventArgs>(string eventName)
    //    {
    //        CreateAsyncEvent(eventName);

    //        return new AsyncTrigger<TEventArgs>(this,eventName);
    //    }

    //    /// <summary>
    //    /// 获取一个同步事件订阅器,可方便的在事件处理里使用,如 <br/>
    //    /// 
    //    /// public event EventHandler&lt;SyncEventArgs&lt;JObject&gt;&gt;  DemoSyncEvent<br/>
    //    /// {<br/>
    //    ///     add{_bus.GeteSyncEventSubsciber("DemoSyncEvent") +=value;}<br/>
    //    ///     remove{_bus.GeteSyncEventSubsciber("DemoSyncEvent") -=value;}<br/>
    //    /// }<br/>
    //    /// 
    //    /// </summary>
    //    /// <typeparam name="T"></typeparam>
    //    /// <param name="eventName"></param>
    //    /// <returns></returns>
    //    public SyncEventSubsciber<T> GeteSyncEventSubsciber<T>(string eventName) where T : SyncEventArgs
    //    {
    //        _subscriberLocker.EnterUpgradeableReadLock();

    //        SyncEventSubsciber<T> subscriber = null;

    //        if (_cacheEventSubsciber.ContainsKey(eventName))
    //        {
    //            subscriber=(SyncEventSubsciber<T>)_cacheEventSubsciber[eventName];
    //        }
    //        else
    //        {
    //            _subscriberLocker.EnterWriteLock();

    //            subscriber = new SyncEventSubsciber<T>(this, eventName);

    //            _cacheEventSubsciber.Add(eventName, subscriber);

    //            _subscriberLocker.ExitWriteLock();
    //        }

    //        _subscriberLocker.ExitUpgradeableReadLock();

    //        return subscriber;
    //    }

    //    /// <summary>
    //    /// 获取一个异步事件订阅器,可方便的在事件处理里使用,如 <br/>
    //    /// 
    //    /// public event EventHandler&lt;AsyncEventArgs&lt;JObject&gt;&gt;  DemoAsyncEvent<br/>
    //    /// {<br/>
    //    ///     add{_bus.GeteSyncEventSubsciber("DemoAsyncEvent") +=value;}<br/>
    //    ///     remove{_bus.GeteSyncEventSubsciber("DemoAsyncEvent") -=value;}<br/>
    //    /// }<br/> 
    //    /// 
    //    /// </summary>
    //    /// <typeparam name="T"></typeparam>
    //    /// <param name="eventName"></param>
    //    /// <returns></returns>
    //    public AsyncEventSubsciber<T> GetAsyncEventSubsciber<T>(string eventName) //where T : EventArgs
    //    {
    //        _subscriberLocker.EnterUpgradeableReadLock();

    //        AsyncEventSubsciber<T> subscriber = null;

    //        if (_cacheEventSubsciber.ContainsKey(eventName))
    //        {
    //            subscriber = (AsyncEventSubsciber<T>)_cacheEventSubsciber[eventName];
    //        }
    //        else
    //        {
    //            _subscriberLocker.EnterWriteLock();

    //            subscriber = new AsyncEventSubsciber<T>(this, eventName);

    //            _cacheEventSubsciber.Add(eventName, subscriber);

    //            _subscriberLocker.ExitWriteLock();
    //        }

    //        _subscriberLocker.ExitUpgradeableReadLock();

    //        return subscriber;
    //    }

    //    /// <summary>
    //    /// 订阅一个事件
    //    /// </summary>
    //    /// <param name="eventName"></param>
    //    public void SubscibeEvent(string[] eventName)
    //    {
    //        subscibeEventRequest(eventName);
    //    }

    //    /// <summary>
    //    /// 退订一个事件
    //    /// </summary>
    //    /// <param name="eventName"></param>
    //    public void UnsubscibeEvent(string[] eventName)
    //    {
    //        unsubscibeEventRequest(eventName);
    //    }

    //    public void Close()
    //    {
    //        var events = _currentSubscibeEvents.ToArray();

    //        if (events.HasData())
    //        {
    //            UnsubscibeEvent(_currentSubscibeEvents.ToArray());

    //            _channel.QueueUnbind(_eventQueue, "RemoteEvent_Brocast", "NodeOnline", null);

    //            foreach (var eventName in events)
    //            {
    //                _channel.QueueUnbind(_eventQueue, "RemoteEvent_EventBus", $"{eventName}_Async_Trigger",null);
    //                _channel.QueueUnbind(_eventQueue, "RemoteEvent_EventBus", $"{eventName}_Sync_Trigger",null);
    //            }

    //        }


    //        _channel.QueueUnbind(_eventQueue, "RemoteEvent_Brocast", "NodeOnline",null);
    //        _channel.QueueUnbind(_eventQueue, "RemoteEvent_Brocast", "NodeSubscibeEvent",null);
    //        _channel.QueueUnbind(_eventQueue, "RemoteEvent_EventBus", $"Queue_{_serviceID}",null);

    //        _channel.Close();
    //        _currentSubscibeEvents.Clear();
    //        _cacheEventSubsciber.Clear();
    //    }

    //    /// <summary>
    //    /// 当一个新服务器节点上线之后
    //    /// </summary>
    //    public event EventHandler<NodeOnlineEventArgs> NodeOnline;

    //    /// <summary>
    //    /// 接收到一个同步事件,在开始处理时(该事件是在订阅端触发)
    //    /// </summary>
    //    public event EventHandler<SyncEventReceiveEventArgs> SyncEventRaising;

    //    /// <summary>
    //    /// 接收到一个同步事件,在完成处理之后(该事件是在订阅端触发的)
    //    /// </summary>
    //    public event EventHandler<SyncEventReceiveEventArgs> SyncEventRaised;

    //    /// <summary>
    //    /// 接收到一个异步事件,在开始处理时(该事件是在订阅端触发)
    //    /// </summary>
    //    public event EventHandler<AsyncEventReceivedEventArgs> AsyncEventRaising;

    //    /// <summary>
    //    /// 接收到一个异步事件,在完成处理之后(该事件是在订阅端触发的)
    //    /// </summary>
    //    public event EventHandler<AsyncEventReceivedEventArgs> AsyncEventRaised;

    //    /// <summary>
    //    /// 当一个订阅队列超时未发送心跳包时,事件触发端会触发该事件
    //    /// </summary>
    //    public event EventHandler<NodeOfflineEventArgs> NodeOffline;


    //    internal IConnection Connection => _conn; 

    //    /// <summary>
    //    /// 当前服务器节点上线
    //    /// </summary>
    //    private void currentNodeOnline()
    //    {
    //        //TODO:向服务器广播通道,广播当前节点上线
    //        /*

    //         */

    //        var prop = createMsgProp();
    //        prop.DeliveryMode = 1;

    //        _channel.BasicPublish("RemoteEvent_Brocast",
    //            "Node_Online",
    //            createMsgProp(),
    //            Encoding.UTF8.GetBytes(
    //                JsonConvert.SerializeObject(new NodeOnlinePackage()
    //                {
    //                    NodeQueue = _eventQueue
    //                }))
    //                );
    //    }

    //    /// <summary>
    //    /// 当前节点订阅事件时,调用该函数向其他节点发送请求
    //    /// </summary>
    //    /// <param name="eventNames"></param>
    //    private void subscibeEventRequest(string[] eventNames)
    //    {
    //        foreach (var eventName in eventNames)
    //        {
    //            if (_currentSubscibeEvents.Add(eventName))
    //            {
    //                _channel.QueueBind(_eventQueue, "RemoteEvent_EventBus", $"{eventName}_Async_Trigger");
    //                _channel.QueueBind(_eventQueue, "RemoteEvent_EventBus", $"{eventName}_Sync_Trigger");

    //            }
    //        }

    //        _channel.BasicPublish("RemoteEvent_Brocast",
    //            "EventSubscibe",
    //            createMsgProp(),
    //            Encoding.UTF8.GetBytes(
    //                JsonConvert.SerializeObject(new NodeSubscibeEventPackage()
    //                {
    //                    EventNames = eventNames,
    //                    SubsciberServiceID = _serviceID
    //                }))
    //                );
    //    }

    //    /// <summary>
    //    /// 向其他服务器通知当前服务器的是活动的,,(定时广播当前服务所订阅的所有事件)
    //    /// </summary>
    //    /// <param name="state"></param>
    //    private void keepNodeAlive(object state)
    //    {
    //        if (_currentSubscibeEvents.Count == 0)
    //        {
    //            return;
    //        }

    //        var events = _currentSubscibeEvents.ToArray();

    //        _channel.BasicPublish("RemoteEvent_Brocast",
    //            "EventSubscibe",
    //            createMsgProp(),
    //            Encoding.UTF8.GetBytes(
    //                JsonConvert.SerializeObject(new NodeSubscibeEventPackage()
    //                {
    //                    EventNames = events,
    //                    SubsciberServiceID = _serviceID
    //                }))
    //                );
    //    }

    //    /// <summary>
    //    /// 当前节点退订事件时,调用该函数向其他节点发送请求
    //    /// </summary>
    //    /// <param name="eventNames"></param>
    //    private void unsubscibeEventRequest(string[] eventNames)
    //    {
    //        foreach (var eventName in eventNames)
    //        {
    //            if (_currentSubscibeEvents.Remove(eventName))
    //            {
    //                _channel.QueueUnbind(_eventQueue, "RemoteEvent_EventBus", $"{eventName}_Async_Trigger", null);
    //                _channel.QueueUnbind(_eventQueue, "RemoteEvent_EventBus", $"{eventName}_Sync_Rely", null);
    //                _channel.QueueUnbind(_eventQueue, "RemoteEvent_EventBus", $"{eventName}_Sync_Trigger", null);

    //                _cacheEventSubsciber.Remove(eventName);
    //            }


    //        }

    //        _channel.BasicPublish("RemoteEvent_Brocast",
    //            "EventUnSubscibe",
    //            createMsgProp(),
    //            Encoding.UTF8.GetBytes(
    //                JsonConvert.SerializeObject(new NodeUnsubscibeEventPackage()
    //                {
    //                    EventNames = eventNames,
    //                    SubsciberServiceID = _serviceID
    //                }))
    //                );
    //    }

    //    private void asyncEventTrigger(string eventName, object data)
    //    {
    //        _channel.BasicPublish("RemoteEvent_EventBus",
    //            $"{eventName}_Async_Trigger",
    //            createMsgProp(),
    //            Encoding.UTF8.GetBytes(
    //                JsonConvert.SerializeObject(new AsyncEventRaisePackage()
    //                {
    //                    Data = JsonConvert.SerializeObject(data),
    //                    EventName = eventName,
    //                    TypeName = data.GetType().FullName,
    //                }))
    //                );
    //    }

    //    /// <summary>
    //    /// 创建一个远程事件触发器时,调用该函数
    //    /// </summary>
    //    /// <param name="eventNames"></param>
    //    private void createEventTrigger(string[] eventNames, bool isAsync)
    //    {
    //        foreach (var name in eventNames)
    //        {
    //            if (!_eventBus.ContainsKey(name) && !isAsync)
    //            {
    //                _eventBus.Add(name, new MemoryCache($"{name}_Subsciber"));
    //            }
    //        }
    //    }

    //    private async Task<JObject[]> syncEventTrigger(string eventName, object data)
    //    {
    //        var triggerID = Guid.NewGuid().ToString();

    //        var queueList = _eventBus.TryGetValue(eventName);

    //        var subsicerlst = new HashSet<string>();

    //        foreach (var pair in queueList)
    //        {
    //            subsicerlst.Add(pair.Key);
    //        }

    //        var item = new SyncEventRelyAwatiable(subsicerlst);

    //        _callback.Add(triggerID, item, new CacheItemPolicy()
    //        {
    //            AbsoluteExpiration = new DateTimeOffset(DateTime.Now).AddSeconds(80),
    //            RemovedCallback = onSyncAwaiterCacheItemRemoved
    //        });

    //        _channel.BasicPublish("RemoteEvent_EventBus",
    //            $"{eventName}_Sync_Trigger",
    //            true,
    //            createMsgProp(),
    //            Encoding.UTF8.GetBytes(
    //                JsonConvert.SerializeObject(new SyncEventRaisePackage()
    //                {
    //                    Data = JsonConvert.SerializeObject(data),
    //                    EventName = eventName,
    //                    TypeName = data.GetType().FullName,
    //                    RelyQueueName = $"Queue_{_serviceID}",
    //                    TriggerID = triggerID
    //                }))
    //                );

    //        try
    //        {
    //            var ret = await item;

    //            return ret.ToArrayEx();

    //            //if (item.GetError() == null)
    //            //{
    //            //    return ret.ToArrayEx();
    //            //}
    //            //else
    //            //{
    //            //    throw item.GetError();
    //            //}
    //        }
    //        catch (Exception ex)
    //        {
    //            onSyncEventError(eventName, data, ex);
    //            throw;
    //        }
    //        finally
    //        {
    //            _callback.Remove(triggerID);
    //        }

    //    }

    //    private void syncEventRely(string eventName, string triggerID, string data, Exception error, string senderQueueName)
    //    {
    //        var package = new SyncEventRelyPackage()
    //        {
    //            EventName = eventName,
    //            ReturnValue = data,
    //            Error = error,
    //            SenderQueueName = _serviceID,
    //            TriggerID = triggerID
    //        };

    //        //var prop = createMsgProp();



    //        _channel.BasicPublish("RemoteEvent_EventBus",
    //                senderQueueName,
    //                createMsgProp(),
    //                Encoding.UTF8.GetBytes(
    //                    JsonConvert.SerializeObject(package)
    //            ));
    //    }


    //    private void Consumer_Received(object sender, BasicDeliverEventArgs e)
    //    {

    //        if (e.Exchange == "RemoteEvent_Brocast")
    //        {
    //            brocastMessageHandler(e);
    //        }
    //        else if (e.Exchange == "RemoteEvent_EventBus")
    //        {
    //            eventMessageHandler(e);
    //        }

    //    }

    //    private bool brocastMessageHandler(BasicDeliverEventArgs e)
    //    {
    //        var routeKey = e.RoutingKey;

    //        if (routeKey.CompareTo("Node_Online", true))
    //        {
    //            if (e.BasicProperties != null)
    //            {
    //                if (e.BasicProperties.AppId == _serviceID)
    //                {
    //                    _channel.BasicAck(e.DeliveryTag, false);
    //                    return false;
    //                }
    //            }

    //            var package = JsonConvert.DeserializeObject<NodeOnlinePackage>(Encoding.UTF8.GetString(e.Body));

    //            onNewNodeOnline(package);
    //        }
    //        else if (routeKey == "EventSubscibe")
    //        {
    //            var package = JsonConvert.DeserializeObject<NodeSubscibeEventPackage>(Encoding.UTF8.GetString(e.Body));

    //            onNodeSubscibeEvent(package.SubsciberServiceID, package.EventNames);

    //            _channel.BasicAck(e.DeliveryTag, false);
    //        }
    //        else if (routeKey == "EventUnSubscibe")
    //        {
    //            var package = JsonConvert.DeserializeObject<NodeUnsubscibeEventPackage>(Encoding.UTF8.GetString(e.Body));

    //            onNodeUnsubscibeEvent(package.SubsciberServiceID, package.EventNames);
    //            _channel.BasicAck(e.DeliveryTag, false);
    //        }


    //        return true;
    //    }

    //    private bool eventMessageHandler(BasicDeliverEventArgs e)
    //    {
    //        var routeKey = e.RoutingKey;

    //        var isSuccess = false;

    //        if (routeKey.EndsWith("_Async_Trigger", StringComparison.CurrentCultureIgnoreCase))
    //        {
    //            _channel.BasicAck(e.DeliveryTag, false);

    //            var package = JsonConvert.DeserializeObject<AsyncEventRaisePackage>(Encoding.UTF8.GetString(e.Body));
    //            //var typeName = package.TypeName;

    //            //var type = Type.GetType(typeName, false) ?? typeof(JObject);

    //            //var value = JsonConvert.DeserializeObject(package.Data, type);


    //            try
    //            {
    //                onAsyncEventTrigger(package.EventName, package.Data);
    //            }
    //            catch (Exception)
    //            {

    //            }

    //            isSuccess = true;
    //        }
    //        else if (routeKey.EndsWith("_Sync_Trigger", StringComparison.CurrentCultureIgnoreCase))
    //        {
    //            var package = JsonConvert.DeserializeObject<SyncEventRaisePackage>(Encoding.UTF8.GetString(e.Body));
    //            var typeName = package.TypeName;

    //            //var type = Type.GetType(typeName, false) ?? typeof(JObject);

    //            //var value = JsonConvert.DeserializeObject(package.Data, type);

    //            try
    //            {
    //                onSyncEventTrigger(package.EventName, package.TriggerID, package.Data, package.RelyQueueName);

    //                _channel.BasicAck(e.DeliveryTag, false);
    //            }
    //            catch (Exception)
    //            {
    //                _channel.BasicNack(e.DeliveryTag, false, true);
    //            }

    //            isSuccess = true;
    //        }
    //        else if (routeKey.StartsWith("Queue_", StringComparison.CurrentCultureIgnoreCase))
    //        {
    //            var json = JObject.Parse(Encoding.UTF8.GetString(e.Body));

    //            var msgType = json.GetString("MessageType");

    //            if (msgType == "SyncEventRely")
    //            {
    //                var package = json.ToObject<SyncEventRelyPackage>();

    //                onSyncEventRely(package.EventName, package.TriggerID, package.ReturnValue, package.SenderQueueName);
    //            }


    //            isSuccess = true;
    //        }

    //        return isSuccess;
    //    }

    //    /// <summary>
    //    /// 当未有客户端订阅该事件时,RabbitMQ会回调BasicReturn事件,触发该函数进行处理,需要在该函数中,直接回调Awaiter的SetResult事件,停止等待
    //    /// </summary>
    //    /// <param name="sender"></param>
    //    /// <param name="e"></param>
    //    private void channel_BasicReturn(object sender, BasicReturnEventArgs e)
    //    {
    //        if (e.RoutingKey.EndsWith("_Sync_Trigger", StringComparison.CurrentCultureIgnoreCase))
    //        {
    //            var package = JsonConvert.DeserializeObject<SyncEventRaisePackage>(Encoding.UTF8.GetString(e.Body));

    //            var item = _callback.Get(package.TriggerID);

    //            if (item!=null)
    //            {
    //                ((SyncEventRelyAwatiable) item).SetResult(new JObject[0], true);
    //            }
    //        }
    //    }


    //    //   事件消息接收后的处理函数_Start

    //    /// <summary>
    //    /// 当其他服务器发送订阅事件的消息时,调用该函数触发事件
    //    /// </summary>
    //    private void onNodeSubscibeEvent(string subsciberQueueName, string[] eventNames)
    //    {
    //        foreach (var eventName in eventNames)
    //        {
    //            MemoryCache lst;//= _eventBus.TryGetValue(eventName);

    //            if (_eventBus.TryGetValue(eventName, out lst))
    //            {
    //                if (lst.Contains(subsciberQueueName))
    //                {
    //                    var s = lst?.Get(subsciberQueueName);
    //                }
    //                else
    //                {
    //                    lst?.Set(subsciberQueueName, 1, new CacheItemPolicy() { SlidingExpiration = new TimeSpan(0, 1, 0), RemovedCallback = onSyncEventSubsciberExpireOffline });
    //                }
    //            }
    //        }
    //    }

    //    /// <summary>
    //    /// 当其他服务器发送退订事件的消息时,调用该函数触发事件
    //    /// </summary>
    //    private void onNodeUnsubscibeEvent(string subsciberQueueName, string[] eventNames)
    //    {
    //        foreach (var eventName in eventNames)
    //        {
    //            var lst = _eventBus.TryGetValue(eventName);

    //            lst?.Remove(subsciberQueueName);
    //        }
    //    }

    //    /// <summary>
    //    /// 当一个事件触发器上线之后的处理(如果上线的事件中,有当前节点需要的事件,则广播一个事件订阅消息)
    //    /// </summary>
    //    /// <param name="eventName"></param>
    //    private void onEventTriggerCreated(string[] eventName)
    //    {

    //    }

    //    /// <summary>
    //    /// 当接收到新的节点上线之后,触发该事件
    //    /// </summary>
    //    private void onNewNodeOnline(NodeOnlinePackage package)
    //    {
    //        /*
    //         * 1.发送当前节点所订阅的事件到新节点
    //         * 
    //         */

    //        NodeOnline?.Invoke(null, new NodeOnlineEventArgs(package.NodeQueue));
    //    }

    //    /// <summary>
    //    /// 当其他服务器触发当前节点订阅的同步事件时,调用该函数触发事件
    //    /// </summary>
    //    private void onSyncEventTrigger(string eventName, string triggerID, string data, string senderQueueName)
    //    {
    //        Exception error = null;

    //        try
    //        {
    //            if (SyncEventRaising != null)
    //            {
    //                var e = new SyncEventReceiveEventArgs()
    //                {
    //                    EventName = eventName,
    //                    Data = data,
    //                    ReturnValue = String.Empty
    //                };

    //                SyncEventRaising?.Invoke(null, e);

    //                data = e.ReturnValue;
    //            }

    //        }
    //        catch (Exception ex)
    //        {
    //            error = ex;
    //        }

    //        syncEventRely(eventName, triggerID, data, error, senderQueueName);

    //        try
    //        {
    //            SyncEventRaised?.Invoke(null, new SyncEventReceiveEventArgs() { EventName = eventName, Data = data });
    //        }
    //        catch (Exception)
    //        {

    //        }

    //    }

    //    /// <summary>
    //    /// 当接收到同步事件远程节点触发完成返回时,调用该函数触发事件
    //    /// </summary>
    //    /// <param name="eventName"></param>
    //    /// <param name="data"></param>
    //    private void onSyncEventRely(string eventName, string triggerID, string data, string senderQueueName)
    //    {
    //        if (_callback.Contains(triggerID))
    //        {
    //            var item = (SyncEventRelyAwatiable)_callback.Get(triggerID);

    //            if (item != null && item.SetResult(senderQueueName, string.IsNullOrWhiteSpace(data) ? null : JObject.Parse(data)))
    //            {
    //                _callback.Remove(triggerID);
    //            }
    //        }
    //    }

    //    /// <summary>
    //    /// 当其他服务器触发当前节点订阅的异步事件时,调用该函数触发事件
    //    /// </summary>
    //    private void onAsyncEventTrigger(string eventName, string data)
    //    {
    //        Exception error = null;

    //        try
    //        {
    //            if (AsyncEventRaising != null)
    //            {
    //                var e = new AsyncEventRaisePackage()
    //                {
    //                    EventName = eventName,
    //                    Data = data
    //                };
    //            }

    //        }
    //        catch (Exception ex)
    //        {
    //            error = ex;
    //        }

    //        try
    //        {
    //            AsyncEventRaised?.Invoke(null, new AsyncEventReceivedEventArgs() { EventName = eventName, Data = data });
    //        }
    //        catch (Exception)
    //        {

    //        }
    //    }

    //    private void onSyncEventError(string eventName, object eventData, Exception error)
    //    {

    //    }

    //    //消息接收后的处理函数_End

    //    private IBasicProperties createMsgProp()
    //    {
    //        return new BasicProperties() { AppId = _serviceID,DeliveryMode = 2};
    //    }

    //    private void onSyncAwaiterCacheItemRemoved(CacheEntryRemovedArguments arguments)
    //    {
    //        if (arguments.RemovedReason == CacheEntryRemovedReason.Expired)
    //        {
    //            var value = (SyncEventRelyAwatiable)arguments.CacheItem.Value;

    //            if (!value.IsCompleted)
    //            {
    //                //TODO:暂时先必须所有已订阅的客户端都必须有回应才可以,下一步可让客户端在订阅时,设定是否强制必须通知到订阅的客户端节点
    //                value.SetError(new TimeoutException($"部分客户端处理超时:队列名称为={value.CheckingList.JoinToString()}"));
    //            }

    //        }
    //    }

    //    /// <summary>
    //    /// 处理当出现事件订阅列表中,有客户端超时未发送心跳包时,触发该函数,引发客户端离线事件
    //    /// </summary>
    //    /// <param name="arguments"></param>
    //    private void onSyncEventSubsciberExpireOffline(CacheEntryRemovedArguments arguments)
    //    {
    //        if (arguments.RemovedReason == CacheEntryRemovedReason.Expired)
    //        {
    //            var eventName = arguments.Source.Name.Split('_')[0];
    //            var subscriberQueue = arguments.CacheItem.Key;
    //            var subscriberID = subscriberQueue.Substring(x => x == '_');

    //            try
    //            {
    //                NodeOffline?.Invoke(this, new NodeOfflineEventArgs(eventName, subscriberID));
    //            }
    //            catch (Exception)
    //            {

    //            }
    //        }
    //    }

    //}
}
